package com.example.iot.controller;

import com.example.iot.actors.ActorSystemContext;
import com.example.iot.entity.dto.RuleChainMetaData;
import com.example.iot.entity.dto.TsData;
import com.example.iot.entity.id.EntityId;
import com.example.iot.entity.id.EntityIdFactory;
import com.example.iot.entity.id.EntityType;
import com.example.iot.entity.id.RuleChainId;
import com.example.iot.msg.*;
import com.example.iot.queue.*;
import com.example.iot.queue.message.RuleEngineQueueMsg;
import com.example.iot.queue.message.ProtoQueueMsg;
import com.example.iot.queue.util.TopicPartitionService;
import com.example.iot.service.core.RuleChainService;
import com.example.iot.service.ts.TimeSeriesService;
import com.example.iot.util.UUIDUtil;
import com.example.iot.websocket.DefaultWebSocketHandler;
import com.example.iot.websocket.WebSocketMsgEndpoint;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.socket.adapter.NativeWebSocketSession;

import javax.websocket.Session;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * <p>
 *  前端控制器
 * </p>
 *
 * @author ChenXiangpeng
 * @since 2021-02-26
 */
@RestController
@RequestMapping("/api/ruleChain")
@Api
public class RuleChainController {
    @Autowired
    private RuleChainService ruleChainService;
    @Autowired
    private ActorSystemContext actorSystemContext;
    @Autowired
    private QueueProducerContext queueProducerContext;
    @Autowired
    private QueueFactory queueFactory;
    @Autowired
    private TopicPartitionService topicPartitionService;

    @GetMapping("/get")
    @ApiOperation("获取节点")
    public RuleChainMetaData getRuleChainMetaData() {
        return ruleChainService.loadRuleChainMetaData(new RuleChainId(UUIDUtil.fromString("51316c99-8bbf-407b-823a-b5648e6ca090")));
    }

    @PostMapping("/save")
    @ApiOperation("保存节点")
    public void saveRuleChainMetaData(@RequestBody RuleChainMetaData ruleChainMetaData) {
        ruleChainService.saveRuleChainMetaData(ruleChainMetaData);
        actorSystemContext.tellWithHighPriority(new ComponentLifecycleMsg(ruleChainMetaData.getRuleChainId(), ComponentLifecycleEvent.UPDATED));
        //ruleNodeService.onUpdate(ruleChainMetaData.getRuleChainId());
    }

    @GetMapping("/test")
    @ApiOperation("测试")
    public void test() {
        Map<String, String> map = Maps.newHashMap();
        map.put("data", "测试设备");
        TbMsg tbMsg = TbMsg.newMsg("Main", "POST_TELEMETRY_REQUEST", EntityIdFactory.getByTypeAndUuid(EntityType.RULE_CHAIN, UUID.randomUUID()), new TbMsgMetaData(map),
                "{\"data\": 22}", (RuleChainId) EntityIdFactory.getByTypeAndUuid(EntityType.RULE_CHAIN, UUID.fromString("51316c99-8bbf-407b-823a-b5648e6ca090")), null);
        QueueToRuleEngineMsg failure = new QueueToRuleEngineMsg(tbMsg, Sets.newHashSet(), "Failure");
        actorSystemContext.tellWithHighPriority(failure);
    }

    @GetMapping("/send")
    @ApiOperation("消息发送")
    public EntityId send() {
        Map<String, String> map = Maps.newHashMap();
        map.put("data", "测试设备");
        TbMsg tbMsg = TbMsg.newMsg("Main", "POST_TELEMETRY_REQUEST", EntityIdFactory.getByTypeAndUuid(EntityType.RULE_CHAIN, UUID.randomUUID()), new TbMsgMetaData(map),
                "{\"data\": 22}", (RuleChainId) EntityIdFactory.getByTypeAndUuid(EntityType.RULE_CHAIN, UUID.fromString("51316c99-8bbf-407b-823a-b5648e6ca090")), null);
        QueueToRuleEngineMsg failure = new QueueToRuleEngineMsg(tbMsg, Sets.newHashSet(), "Failure");

        RuleEngineQueueMsg build = new RuleEngineQueueMsg(tbMsg, null, null);
        queueProducerContext.getRuleEngineProducer().send(topicPartitionService.resolve("rule_engine.main", tbMsg.getOriginator())
                , new ProtoQueueMsg<>(tbMsg.getId(), build), new QueueCallback() {
            @Override
            public void onSuccess(QueueMsgMetadata metadata) {
                System.out.println(metadata);
            }

            @Override
            public void onFailure(Throwable t) {
                t.printStackTrace();
            }
        });
        return tbMsg.getOriginator();
    }


    @PostMapping("/queue/list")
    @ApiOperation("消费消息")
    public void getList(@RequestBody List<EntityId> entityIdList){
        QueueConsumer<ProtoQueueMsg<RuleEngineQueueMsg>> ruleEngineConsumer = queueFactory.createRuleEngineConsumer();
        ruleEngineConsumer.subscribe(entityIdList.stream().map(e-> topicPartitionService.resolve("rule_engine.main", e)).collect(Collectors.toSet()));
        System.out.println(ruleEngineConsumer.poll(25));
    }

    @Autowired
    private WebSocketMsgEndpoint webSocketMsgEndpoint;
    @PostMapping("/websocket/send/{id}")
    @ApiOperation("WebSocket发送消息")
    public void sendWebsocket(@PathVariable String id){
        webSocketMsgEndpoint.send(id, "测试消息");
    }
    @Autowired
    private TimeSeriesService timeSeriesService;

    @PostMapping("/device/{deviceId}")
    @ApiOperation("保存设备数据")
    public void saveDeviceData(@PathVariable String deviceId, @RequestBody TsData data){
        timeSeriesService.save(deviceId, data);
    }

}
